package com.haozhen.sql

/**
  * @author haozhen
  * @email haozh@ync1.com
  * @date 2021/1/31  18:19
  */
object EmpDemo {

  def main(args: Array[String]): Unit = {
    import org.apache.spark.sql.expressions.WindowSpec
    import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
    val spark: SparkSession = SparkSession.builder().master("local[*]").appName(this.getClass.getCanonicalName().init).getOrCreate()
    val sc = spark.sparkContext
    sc.setLogLevel("WARN")

    import spark.implicits._
    import org.apache.spark.sql.functions._
    val df1: DataFrame = spark.read.option("header", "true")
      .option("inferschema", "true")
      .csv("data/emp.dat")
    val count: Long = df1.count()

//    df1.union(df1).show(100)

//    df1.show(2)

//    df1.toJSON.show(false)

//    df1.toJSON.show(10,false)


//    val array: Array[Dataset[Row]] = df1.randomSplit(Array(0.5,0.6,0.7))
//
//    array(0).show()
//    array(1).show()
//    array(2).show()

//    val df2: Dataset[Row] = df1.union(df1)
//    df2.distinct().count()
//
//    df2.dropDuplicates().show()
//
//    df2.dropDuplicates("mgr","deptno").show()
//
//    df2.dropDuplicates("mgr").show()
//
//    df2.dropDuplicates("deptno").show()
//
//
//    df2.describe().show()

//    df1.select("empno").show()
//
//    df1.select($"empno",$"sal").show()
//
//    df1.select('empno,'sal).show()
//
//    df1.select(col("empno"),col("sal")).show()
//
//    df1.select(df1("empno"),df1("sal")).show()
//
//    df1.select(expr("sal+1000")).show()
//
//    df1.selectExpr("ename as name").show()
//
//    //修改列的定义
//    df1.withColumn("sal",$"sal"+1000).show()
//    //基于原有的列曾加新列
//    df1.withColumn("newsal",$"sal").show()
//    //修改列名
//    df1.withColumnRenamed("sal","sal2").printSchema()
//    //修改列的类型
//    df1.selectExpr("cast(empno as string)").printSchema()
//
//
//    //升序
//    df1.orderBy($"sal".asc).show
//
//    //降序
//    df1.orderBy(-$"sal").show
//
//
//    //空值处理
//    math.sqrt(-1.0).isNaN
//    df1.na.fill(0).show()
//    df1.na.drop(List("comm")).show()
//    df1.na.fill(Map("mgr"->9999,"comm"->0)).na.replace(List("mgr","comm"),Map(9999->6666,0->100))
//    df1.where($"mgr".isNotNull).show()

    import org.apache.spark.sql.expressions.Window



//    val w1 = Window.partitionBy("cookieod").orderBy("createtime")
//    val w2 = Window.partitionBy("cookieid").orderBy("pv")
//    val w3: WindowSpec = w1.rowsBetween(Window.unboundedPreceding,Window.currentRow)
//    val w4: WindowSpec = w1.rowsBetween(-1,1)
//
//
//    //聚合函数
//    df1.select($"cookieid",$"pv",sum("pv").over(w1).alias("pv1")).show()
//
//    df1.select($"cookieid",$"pv",sum("pv").over(w3).alias("pv2")).show()
//
//    //排名函数
//    df1.select($"cookieid",$"pv",rank().over(w2).alias("rank")).show()
//    df1.select($"cookieid",$"pv",dense_rank().over(w3).alias("dense_rank")).show()
//    df1.select($"cookieid",$"pv",row_number().over(w3).alias("row_number")).show()
//    //lag lead
//    df1.select($"cookieid",$"pv",lag("pv",2).over(w2).alias("pv1")).show()
//    df1.select($"cookieid",$"pv",lag("pv",-2).over(w2).alias("pv1")).show()

    df1.createTempView("emp");
//    df1.createOrReplaceTempView("emp2")
//    df1.createGlobalTempView()
    spark.sql(
      """
        |select * from emp
      """.stripMargin).show()
    spark.close()
  }
}
